package practices; /**
 * @Description @TODO
 * @Date 2025/4/15 11:05
 * @Created by weimeiling
 */

import java.util.LinkedList;
import java.util.Queue;
import java.util.Random;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.ForkJoinPool;
import java.util.concurrent.RecursiveTask;

/**
 ** 原理
 * 构造一个 Thread 类，调用start()方法启动线程
 * 并在 run() 中，定义需要执行的任务
 *
 *
 ** 实现线程的方式
 * 1、实现 Runable 接口实现线程
 * 2、继承 Thread 类
 * 3、通过线程池创建线程
 * 4、通过有返回值的 callable 创建线程
 * 5、定时器 Timer class TimerThread extends Thread
 * 6、匿名内部类
 * 7、Lambda 表达式
 *
 *
 ** 透过现象看本质，最终的实现都是实现 runnable 接口或者继承 Thread 类
 ** 创建线程的唯一方法：构造 Thread 类
 *
 *
 ** 总结
 * 1、实现runnable接口要比继承thread类要更好
 * 可以把不同的内容进行解耦，权责分明
 * 某些情况下可以提升性能，减小开销
 * 继承 Thread 类相当于限制了代码未来的可扩展性
 *
 * -----------------------------------------------------------------
 * 优雅的停止线程 interrupt
 *         while (!Thread.currentThread().isInterrupted() && more work to do) {
 *             do more work
 *         }
 *  demo StopThread
 *
 *  掌握了为什么说 volatile 修饰标记位停止方法是错误的
 *  volatile 这种方法在某些特殊的情况下，比如线程被长时间阻塞的情况，就无法及时感受中断，所以volatile 是不够全面的停止线程的方法
 */

public class MyRunnable {
    public static void main(String[] args) {
        //匿名内部类
        new Thread(new Runnable() {
            @Override
            public void run() {
                System.out.println("第六种：匿名内部类");
                System.out.println(Thread.currentThread().getName());
            }
        }).start();

        //
        new Thread(()->{
            System.out.println("第六种：Lambda 表达式");
            System.out.println(Thread.currentThread().getName());
        }).start();

        // 优雅的停止线程 interrupt

    }
}

class ImplementRunnable implements Runnable {
    //实现Runnable接口的run 方式，并把Runnable实例作为target对象，传给Thread类，最终调用target.run（）
    @Override
    public void run() {
        System.out.println("第一种：实现 Runable 接口实现线程");
    }
}

class ExtendsThread extends Thread {
    //继承Thread类，重写Thread的run方法（），Thread.start （） 会执行run（）
    @Override
    public void run() {
        System.out.println("第二种：继承 Thread 类实现线程");
    }
}

class ThreadFactory {
    public ExecutorService createThread(){
        // 创建一个固定大小为 3 的线程池
        System.out.println("第三种：通过线程池创建线程");
        ExecutorService executor = Executors.newFixedThreadPool(3);
//        executor.submit(new CallableTask());
        return executor;
    }
}

class CallableTask implements Callable<Integer>{

    @Override
    public Integer call() throws Exception {
        System.out.println("第四种：通过有返回值的 callable 创建线程");
        return new Random().nextInt();
    }
}

class StopThread implements Runnable{

    @Override
    public void run() {
        int count = 0;
        while (!Thread.currentThread().isInterrupted() && count < 1000){
            System.out.println("count = " + count++);
            if (count == 500){
                Thread.currentThread().interrupt();
            }
        }
    }

    public static void main(String[] args) throws InterruptedException {
        Thread thread = new Thread(new StopThread());
        thread.start();

        Thread.sleep(20);
        thread.interrupt();
    }
}


/**
 ** 线程的六种状态
 * NEW 新创建的线程，尚未启动
 * RUNNABLE 可运行的线程，正在运行或者等待CPU调度
 * SLEEPING 休眠状态，线程不会被分配CPU时间，到了指定的时间会自动恢复
 *
 * WAITING 等待状态，线程不会被分配CPU时间，需要其他线程显式地唤醒
 * TIMED_WAITING 超时等待状态，线程不会被分配CPU时间，到了指定的时间会自动恢复
 * BLOCKED 阻塞状态，线程被阻塞，需要其他线程释放锁
 *
 * TERMINATED 终止状态，线程已经执行完毕
 *
 */


/**
 * 为什么 wait 必须在 synchronized 保护的同步代码中使用
 * 在 Java 里，wait() 方法是 Object 类的一部分，必须在 synchronized 保护的同步代码块或者同步方法中使用，主要有以下几方面原因：
 * 1. 保证线程安全
 * wait() 方法的主要用途是让当前线程进入等待状态，直至其他线程调用相同对象的 notify() 或者 notifyAll() 方法。若不在同步代码块中使用 wait()，就可能出现竞态条件。
 * 比如，当一个线程检查某个条件（像队列是否为空），发现条件不满足，准备调用 wait() 时，另一个线程可能会在这个线程调用 wait() 之前修改该条件（例如往队列中添加元素），然后调用 notify()。这样一来，调用 wait() 的线程就会错过这个通知，从而陷入无限等待。
 * 2. 与锁的状态关联
 * wait() 方法在执行时会释放对象的锁。这意味着，当一个线程调用 wait() 后，其他线程就能够获取这个对象的锁并执行同步代码块。要是不在同步代码块中调用 wait()，由于没有锁可释放，就会抛出 IllegalMonitorStateException 异常。
 * 3. 确保通知和等待的一致性
 * notify() 和 notifyAll() 方法也必须在同步代码块中调用，目的是确保在通知等待线程时，对象的状态是一致的。在同步代码块中使用 wait()、notify() 和 notifyAll()，能够保证线程之间的通信是有序且安全的。
 */
class SharedObject {
    private boolean flag = false;

    public synchronized void waitForFlag() throws InterruptedException {
        // 检查条件
        while (!flag) {
            // 条件不满足，当前线程进入等待状态
            wait();
        }
        System.out.println("Flag is set, proceeding...");
    }

    public synchronized void setFlag() {
        // 设置标志位
        flag = true;
        // 通知所有等待的线程
        notifyAll();
    }
}

 class WaitNotifyExample {
    public static void main(String[] args) {
        SharedObject sharedObject = new SharedObject();

        // 创建一个等待线程
        Thread waitingThread = new Thread(() -> {
            try {
                sharedObject.waitForFlag();
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
        });

        // 创建一个设置标志位的线程
        Thread settingThread = new Thread(() -> {
            try {
                // 模拟一些工作
                Thread.sleep(1000);
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
            sharedObject.setFlag();
        });

        // 启动线程
        waitingThread.start();
        settingThread.start();
    }
}


/**
 * 如何用wait/notify 实现生产者消费者模式
 */


// ... existing code ...

class Buffer {
    private final Queue<Integer> queue = new LinkedList<>();
    private final int MAX_SIZE = 5;
    public synchronized void produce(int item) throws InterruptedException {
        while (queue.size() == MAX_SIZE) {
            System.out.println("缓冲区已满，生产者 " + Thread.currentThread().getName() + " 等待...");
            wait(); // 缓冲区已满，生产者等待
        }
        queue.add(item);
        System.out.println("生产者 " + Thread.currentThread().getName() + " 生产了, item: " + item + ", 当前缓冲区大小: " + queue.size());
        notifyAll(); // 通知消费者可以消费
    }

    public synchronized void consume() throws InterruptedException {
        while (queue.size() == 0) {
            System.out.println("缓冲区为空，消费者 " + Thread.currentThread().getName() + " 等待...");
            wait(); // 缓冲区为空，消费者等待
        }
        Integer item = queue.poll();
        System.out.println("消费者 " + Thread.currentThread().getName() + " 消费了, item: " + item + ", 当前缓冲区大小: " + queue.size());
        notifyAll(); // 通知生产者可以生产
    }
}

class Producer implements Runnable {
    private final Buffer buffer;

    public Producer(Buffer buffer) {
        this.buffer = buffer;
    }

    @Override
    public void run() {
        System.out.println("生产者 " + Thread.currentThread().getName() + " 开始工作");
        for (int i = 0; i < 10; i++) {
            try {
                buffer.produce(i);
            } catch (InterruptedException e) {
                System.out.println("生产者 " + Thread.currentThread().getName() + " 被中断");
                Thread.currentThread().interrupt();
            }
        }
        System.out.println("生产者 " + Thread.currentThread().getName() + " 工作结束");
    }
}

class Consumer implements Runnable {
    private final Buffer buffer;

    public Consumer(Buffer buffer) {
        this.buffer = buffer;
    }

    @Override
    public void run() {
        System.out.println("消费者 " + Thread.currentThread().getName() + " 开始工作");
        for (int i = 0; i < 10; i++) {
            try {
                buffer.consume();
                Thread.sleep(200);
            } catch (InterruptedException e) {
                System.out.println("消费者 " + Thread.currentThread().getName() + " 被中断");
                Thread.currentThread().interrupt();
            }
        }
        System.out.println("消费者 " + Thread.currentThread().getName() + " 工作结束");
    }
}

class ProducerConsumerExample {
    public static void main(String[] args) {
        System.out.println("主程序启动，开始创建生产者和消费者线程");
        Buffer buffer = new Buffer();
        Thread producerThread = new Thread(new Producer(buffer), "ProducerThread");
        Thread consumerThread = new Thread(new Consumer(buffer), "ConsumerThread");

        producerThread.start();
        consumerThread.start();

        try {
            System.out.println("主程序等待生产者和消费者线程执行完毕");
            producerThread.join();
            consumerThread.join();
        } catch (InterruptedException e) {
            System.out.println("主程序在等待过程中被中断");
            Thread.currentThread().interrupt();
        }
        System.out.println("主程序结束，生产者和消费者线程已执行完毕");
    }
}

/**
 * 什么是线程安全
 * 线程安全是指在多线程环境下，程序的执行结果与单线程执行的结果相同。
 * 换句话说，线程安全是指在多线程环境下，程序的执行不会出现错误或不一致的结果。
 * 线程安全的实现方法有以下几种：
 * 1. 同步方法：使用 synchronized 关键字修饰方法，确保同一时间只有一个线程可以执行该方法。
 * 2. 同步代码块：使用 synchronized 关键字修饰代码块，确保同一时间只有一个线程可以执行该代码块。
 * 3. 不可变对象：如果一个对象在创建后其状态就不能被修改，那么它就是线程安全的。
 * 4. 线程局部变量：使用 ThreadLocal 类来存储线程局部变量，每个线程都有自己的变量副本，不会出现线程安全问题。
 * 5. 无状态对象：如果一个对象没有内部状态，即它的行为只依赖于输入参数，那么它就是线程安全的。
 * 6. 不可变对象：如果一个对象在创建后其状态就不能被修改，那么它就是线程安全的。
 * 7. 原子操作：使用原子类来实现线程安全的操作，例如 AtomicInteger、AtomicBoolean 等。
 * 8. 锁分离：将锁的粒度分离，每个对象都有自己的锁，这样可以提高并发性能。
 * 9. 读写锁：使用读写锁来实现线程安全的读写操作，读操作可以并发执行，写操作需要互斥执行。
 * 10. 线程安全的集合类：使用线程安全的集合类来存储数据，例如 ConcurrentHashMap、CopyOnWriteArrayList 等。
 * 11. 线程安全的容器类：使用线程安全的容器类来存储数据，例如 Vector、Hashtable 等。
 * 12. 线程安全的工具类：使用线程安全的工具类来实现线程安全的操作，例如 Collections.synchronizedList、Collections.synchronizedMap 等。
 * 13. 线程安全的类：使用线程安全的类来实现线程安全的操作，例如 StringBuffer、StringBuilder 等。
 */


/**
 * 线程池拒绝策略
 * 1. AbortPolicy：默认的拒绝策略，抛出 RejectedExecutionException 异常。
 * 2. DiscardPolicy：直接丢弃任务，不抛出异常。
 * 3. DiscardOldestPolicy：丢弃队列中最老的任务，然后尝试重新提交当前任务。
 * 4. CallerRunsPolicy：由调用线程处理该任务。
 * 5. 自定义拒绝策略：实现 RejectedExecutionHandler 接口，自定义拒绝策略。
 */

/**
 * 六种常见的线程池
 * 1. FixedThreadPool：固定大小的线程池，核心线程数和最大线程数相等，队列使用无界队列 LinkedBlockingQueue。
 * 2. CachedThreadPool：可缓存的线程池，核心线程数为 0，最大线程数为 Integer.MAX_VALUE，队列使用 SynchronousQueue。
 * 3. SingleThreadExecutor：单线程线程池，只有一个核心线程，最大线程数为 1，队列使用无界队列 LinkedBlockingQueue。
 * 4. ScheduledThreadPool：定时任务线程池，支持定时和周期性执行任务，队列使用 DelayQueue。
 * 5. WorkStealingPool：工作窃取线程池，使用 ForkJoinPool 实现，适合执行 CPU 密集型任务。
 * 6. ThreadPoolExecutor：自定义线程池，允许用户指定线程池的参数，如核心线程数、最大线程数、队列类型等。
 * 7. ForkJoinPool：ForkJoin 线程池，用于执行 ForkJoinTask 任务，适合执行 CPU 密集型任务。 golang 的协程
 */

/**
 * ForkJoinPool的使用 API 介绍
 * 1. submit(ForkJoinTask task)：提交一个 ForkJoinTask 任务。
 * 2. invoke(ForkJoinTask task)：提交一个 ForkJoinTask 任务，并等待任务完成。
 * 3. execute(Runnable task)：提交一个 Runnable 任务。
 * 4. shutdown()：关闭线程池，不再接受新的任务。
 * 5. shutdownNow()：关闭线程池，尝试停止所有正在执行的任务。
 * 6. isTerminated()：判断线程池是否已终止。
 * 7. isShutdown()：判断线程池是否已关闭。
 * 8. awaitTermination(long timeout, TimeUnit unit)：等待线程池终止，最多等待指定的时间。
 * 9. getPoolSize()：获取线程池中的线程数。
 * 10. getActiveThreadCount()：获取线程池中的活动线程数。
 * 11. getParallelism()：获取线程池中的并行级别。
 * 12. getCommonPoolParallelism()：获取 ForkJoinPool 的并行级别。
 * 13. getCommonPool()：获取 ForkJoinPool 的实例。
 * 
 * 
 * ForkJoinPool 的工作原理
 * 1. ForkJoinPool 是一个线程池，用于执行 ForkJoinTask 任务。
 * 2. ForkJoinTask 是一个抽象类，有两个子类：RecursiveAction 和 RecursiveTask。
 * 3. RecursiveAction 用于执行没有返回值的任务，RecursiveTask 用于执行有返回值的任务。
 * 4. ForkJoinPool 中的工作线程称为工作窃取线程，它们从队列中获取任务并执行。
 * 5. 工作窃取线程从队列中获取任务时，如果队列为空，它们会从其他工作窃取线程的队列中窃取任务。
 * 6. 工作窃取线程执行任务时，如果任务比较大，它们会将任务拆分成更小的子任务，并将子任务添加到队列中。
 * 7. 工作窃取线程执行任务时，如果任务比较小，它们会直接执行任务。
 * 8. 工作窃取线程执行任务时，如果任务执行完毕，它们会将任务从队列中移除。
 * 9. 工作窃取线程执行任务时，如果任务执行过程中出现异常，它们会将任务从队列中移除，并将异常信息记录下来。
 * 
 * 
 * ForkJoinPool 的使用注意事项
 * 1. ForkJoinPool 中的工作线程是守护线程，当所有的非守护线程都结束时，工作线程会自动结束。
 * 2. ForkJoinPool 中的工作线程是协作式的，它们不会主动放弃 CPU 时间片，而是会根据任务的优先级和队列的状态来决定是否放弃 CPU 时间片。
 * 3. ForkJoinPool 中的工作线程是线程安全的，它们可以安全地共享数据。
 * 4. ForkJoinPool 中的工作线程是可重用的，它们可以重复使用，避免了频繁地创建和销毁线程的开销。
 * 5. ForkJoinPool 中的工作线程是并行的，它们可以同时执行多个任务，提高了程序的执行效率。
 * 6. ForkJoinPool 中的工作线程是可配置的，用户可以根据自己的需求来配置线程池的参数，如核心线程数、最大线程数、队列类型等。
 * 7. ForkJoinPool 中的工作线程是可扩展的，用户可以根据自己的需求来扩展线程池的大小，以适应不同的任务。
 * 8. ForkJoinPool 中的工作线程是可监控的，用户可以通过监控线程池的状态来了解线程池的运行情况。
 * 
 * ForkJoinPool 的企业应用场景及伪代码示例
 * 1. 并行计算：ForkJoinPool 可以用于执行并行计算任务，例如矩阵乘法、快速排序等。
 * 2. 大数据处理：ForkJoinPool 可以用于处理大数据集，例如搜索引擎、数据分析等。
 * 3. 分布式计算：ForkJoinPool 可以用于执行分布式计算任务，例如分布式文件系统、分布式数据库等。 
 */

class ForkJoinExample {
    public static void main(String[] args) {
        // 创建一个 ForkJoinPool
        ForkJoinPool pool = new ForkJoinPool();
    }
}

/**
 *  ForkJoinPool 计算数组元素之和
 */

class SumTask extends RecursiveTask<Integer> {
    private static final int THRESHOLD = 10;
    private final int[] array;
    private final int start;
    private final int end;

    public SumTask(int[] array, int start, int end) {
        this.array = array;
        this.start = start;
        this.end = end;
    }

    @Override
    protected Integer compute() {
        if (end - start <= THRESHOLD) {
            // 任务足够小，直接计算
            int sum = 0;    
            for (int i = start; i < end; i++) {
                sum += array[i];
            }
            return sum;
        } else {
            // 任务太大，拆分成两个子任务
            int mid = (start + end) / 2;
            SumTask leftTask = new SumTask(array, start, mid);
            SumTask rightTask = new SumTask(array, mid, end);
            leftTask.fork();
            rightTask.fork();
            
            int leftRuslt = leftTask.join();
            int rightRuslt = rightTask.join(); // 等待任务执行完毕并返回结果
            /**
             * fork() 方法用于将任务提交到线程池，
             * compute() 方法用于计算任务的结果，
             * join() 方法用于等待任务执行完毕并返回结果。
             * fork、compute、join 三个方法的执行顺序如下：
             * 1. fork() 方法用于将任务提交到线程池，
             * 2. compute() 方法用于计算任务的结果，
             * 3. join() 方法用于等待任务执行完毕并返回结果。
             */

            return leftRuslt + rightRuslt;
        }
    }
}

class ForkJoinPoolExample {
    public static void main(String[] args) {
        int[] array = new int[100];
        for (int i = 0; i < 100; i++) {
            array[i] = i + 1;
        }
        // 创建一个 ForkJoinPool
        ForkJoinPool pool = new ForkJoinPool();
        SumTask task = new SumTask(array, 0, array.length);

        // 提交任务并获取结果
        int result = pool.invoke(task);
        System.out.println("Sum: " + result);
    }
}

/**
 * 合适的线程数量是多少，CPU 核心数和线程数的关系？
 * CPU 密集型任务：线程数 = CPU 核心数 + 1
 * IO 密集型任务：线程数 = CPU 核心数 * (1 + IO 等待时间 / CPU 计算时间)
 * 混合型任务：线程数 = CPU 核心数 * (1 + 任务等待时间 / 任务计算时间)
 */

/**
 * 如何正确关闭线程池  使用协同信号（interrupt），关闭
 * 1. shutdown()：关闭线程池，不再接受新的任务。
 * 2. shutdownNow()：关闭线程池，尝试停止所有正在执行的任务。
 * 3. isTerminated()：判断线程池是否已终止。
 * 4. isShutdown()：判断线程池是否已关闭。
 * 5. awaitTermination(long timeout, TimeUnit unit)：等待线程池终止，最多等待指定的时间。
 */


